/**
 * Copyright 2009-2010 by The Regents of the University of California
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * you may obtain a copy of the License from
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package edu.ucr.cs.dblab.acestore.api;

import java.nio.ByteBuffer;

import edu.ucr.cs.dblab.acestore.base.datatypes.SchemaDescriptor;

/**
 * 
 */
public class QuickSorter {

	private static final int INTSIZE = Integer.SIZE / 8;

	public static void Quicksort(ByteBuffer page,
			SchemaDescriptor schemaDescriptor, int[] compareFields,
			int low, int high) throws Exception {
		if (high > low) {
			int partitionPivotIndex = (int) (Math.random() * (high - low)) / 2
					* 2 + low;
			int newPivotIndex = partition(page, schemaDescriptor,
					compareFields, low, high, partitionPivotIndex);
			Quicksort(page, schemaDescriptor, compareFields, low,
					newPivotIndex - 2);
			Quicksort(page, schemaDescriptor, compareFields, 
					newPivotIndex + 2, high);
		}
	}

	private static int partition(ByteBuffer page,
			SchemaDescriptor schemaDescriptor, int[] compareFields,
			int lo, int hi, int pivotIndex) throws Exception {

		swap(page, pivotIndex, hi); // send to the back

		int index = lo;

		for (int i = lo; i < hi; i = i + 2) {
			if (compareLinks(page, schemaDescriptor, compareFields, i,
					hi) <= 0) {
				swap(page, i, index);
				index = index + 2;
			}
		}

		swap(page, hi, index);

		return index;
	}

	private static void swap(ByteBuffer page, int i, int j) {
		int offset1 = PageAccessHelper.GetRecStartOffset(page, i);
		int offset2 = PageAccessHelper.GetRecStartOffset(page, j);
		int len1 = PageAccessHelper.GetRecEndOffset(page, i) + 1 - offset1;
		int len2 = PageAccessHelper.GetRecEndOffset(page, j) + 1 - offset2;
		
		// Put record 2 into a temporary byte array
		byte[] tmp = new byte[len2];
		page.position(offset2);
		page.get(tmp);
		
		// Move record 1 to the position of record 2
		page.position(offset2);
		page.put(page.array(), offset1, len1);
		
		// Move record 2 from the temporary array to the position of record 1
		page.position(offset1);
		page.put(tmp);
	}

	private static int compareLinks(ByteBuffer page,
			SchemaDescriptor schemaDescriptor, int[] compareFields,
			int i, int j) throws Exception {
		return compareRecords(page, schemaDescriptor, compareFields,
				i, j);
	}

	private static int compareRecords(ByteBuffer page,
			SchemaDescriptor schemaDescriptor, int[] compareFields, int idx1,
			int idx2) throws Exception {
		int offset1 = PageAccessHelper.GetRecStartOffset(page, idx1);
		int offset2 = PageAccessHelper.GetRecStartOffset(page, idx2);
		int fieldCount1 = page.getInt(offset1);
		int fieldCount2 = page.getInt(offset2);
		for (int i = 0; i < compareFields.length; i++) {
			int len = schemaDescriptor.getFields()[compareFields[i]]
					.getTypeSize();

			int cmp = schemaDescriptor.getFields()[compareFields[i]].compare(
					page.array(), offset1 + INTSIZE + fieldCount1 * INTSIZE
							+ compareFields[i] * INTSIZE, len, page.array(),
					offset2 + INTSIZE + fieldCount2 * INTSIZE
							+ compareFields[i] * INTSIZE, len);

			if (cmp == 0)
				continue;
			else if (cmp > 0)
				return 1;
			else
				return -1;
		}
		return 0;
	}

}
